package com.intelli.eve.core.impl.message_bus;

import com.intelli.eve.Eve;
import com.intelli.eve.core.api.message.Message;
import com.intelli.eve.core.api.message.Receiver;
import com.intelli.eve.core.impl.configuration.Configuration;
import com.intelli.eve.core.impl.threads.BackgroundTaskManager;
import com.intelli.eve.core.meta.ContextListener;
import com.intelli.eve.core.meta.SupportDisable;
import com.intelli.ray.core.Inject;
import com.intelli.ray.core.ManagedComponent;

import javax.annotation.PostConstruct;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Author: Sergey42
 * Date: 21.12.2014 19:48
 */
@ManagedComponent(name = "MessageDispatcher")
public class MessageDispatcher implements SupportDisable {

    @Inject
    protected MessageBus bus;
    @Inject
    protected BackgroundTaskManager backgroundTaskManager;
    @Inject
    protected Configuration configuration;

    protected volatile boolean enable = false;
    protected ExecutorService executorService;
    protected Map<String, Receiver> receivers = new ConcurrentHashMap<>();

    @PostConstruct
    protected void init() {
        int poolSize = configuration.getIntProperty("messageDispatcher.poolSize", 4);
        executorService = Executors.newFixedThreadPool(poolSize);
        Eve.addListener(new ContextListener() {
            @Override
            public void onStart() {
                registerReceivers();
                backgroundTaskManager.submitTask(new DispatchRunnable());
                enable = true;
            }
        });
    }

    protected void registerReceivers() {
        Collection<Receiver> beansByType = Eve.getContext().getBeansByType(Receiver.class);
        for (Receiver receiver : beansByType) {
            receivers.put(receiver.getProtocol(), receiver);
        }
    }

    @Override
    public void setEnable(boolean on) {
        enable = on;
    }

    protected class DispatchRunnable implements Runnable {

        @SuppressWarnings("all")
        @Override
        public void run() {
            while (true) {
                if (enable) {
                    final Message message = bus.poll();
                    if (message == null) continue;
                    String protocol = message.getHeader().getProtocol();
                    if (protocol == null) continue;
                    final Receiver receiver = receivers.get(protocol);
                    if (receiver == null) continue;
                    executorService.submit(new Runnable() {
                        @Override
                        public void run() {
                            receiver.handle(message);
                        }
                    });
                }
            }
        }
    }
}
